/*
 * @Company: TWL
 * @Author: xue jian
 * @Email: xuejian@kanzhun.com
 * @Date: 2020-02-04 18:20:43
 */
#include "test.h"
#include <thread>
void scan_active_service(std::vector<ActItemThrift>& all_d) {
    std::shared_ptr<TSocket> socket(new TSocket("172.21.34.176", 50880));
    socket->setRecvTimeout(35000);
    // std::shared_ptr<TTransport> transport(new TFramedTransport(socket));//for the noblockingserver
    std::shared_ptr<TTransport> transport(new TBufferedTransport(socket));//for the simpleserver

    std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
    std::shared_ptr<TMultiplexedProtocol> mpl(new TMultiplexedProtocol(protocol, "GeekExpectActService"));
    // std::shared_ptr<TMultiplexedProtocol> mpl(new TMultiplexedProtocol(protocol, "BossJobActService"));

    //RegistClient client(protocol);

    transport->open();

    std::shared_ptr<GeekExpectActServiceThriftClient> client = std::make_shared<GeekExpectActServiceThriftClient>(mpl);


    std::string session_id;
    client->getSessionId(session_id);
    ActResultSetThrift res_set;

    while(true){
        client->scan(res_set, session_id, BusinessTypeThrift::BOSS);
        std::vector<ActItemThrift>& d = res_set.data;
        if(d.empty()){
            continue;
        }
        all_d.insert(all_d.end(), d.begin(), d.end());

        if(!res_set.hasMore){
            break;
        }
        // if (all_d.size()>1000000) {
        //     break;
        // }

    }

    transport->close();
}


void thread_deal(const std::vector<ActItemThrift>& all_d, const size_t& start, const size_t& end) {
    std::string zk_url = "172.21.32.168:2181,172.21.32.136:2181,172.21.32.113:2181,172.21.32.116:2181,172.21.32.196:2181";
    std::string zk_path = "/algorithm_research_center/group6/hulk_proxy/wordbag";
    std::shared_ptr<tcmcp_HulkProxyServClient> client(new tcmcp_HulkProxyServClient(zk_url, zk_path, 1, 100000));
    if (client->init()) {
        std::cout<<"init the client fail ... "<<std::endl;
        exit(1);
    }
    std::map<std::string, std::set<std::string>> hash_keys;
    std::map<std::string, std::map<std::string, std::string>> hash_key_vals;
    size_t success_count(0), fail_count(0);

    std::cout<<"all d size is "<<all_d.size()<<std::endl;
    for (size_t i(start); i<end; ++i) {
        std::string hash("ewb:0:");
        hash += std::to_string(all_d.at(i).jepId);
        std::string key("3_0");
        std::set<std::string> real_key;
        real_key.insert(key);
        hash_keys.insert(std::make_pair(hash, real_key));
        if (i%10000 == 0) {
            client->tcmcp_mhmget(hash_key_vals, hash_keys);
            hash_keys.clear();
            for (auto&hash_key_val:hash_key_vals) {
                for (auto&key_val:hash_key_val.second) {
                    try{
                        json word_j = json::parse(key_val.second);
                        success_count++;
                    } catch (...) {
                        // auto len = key_val.second.length() - 1;
                        // for ( ; len>=0; len--) {
                        //     if (key_val.second.at(len) == '}') {
                        //         break;
                        //     };
                        // }
                        size_t len = key_val.second.length();
                        if (key_val.second.at(len-1) == '}'){
                            len--;
                        }
                        for ( ; len>=1; len--) {
                            if (key_val.second.at(len-1) == '}') {
                                break;
                            };
                        }
                        key_val.second = key_val.second.substr(0, len);

                        // std::cout<<"The error word_bag is "<<it_word->second<<std::endl;
                        fail_count++;
                        continue;
                    }
                }
            }
            client->tcmcp_mhmpush(hash_key_vals);
            hash_key_vals.clear();
            std::cout<<"The success number is "<<success_count<<". The fail number is "<<fail_count<<std::endl;
        }
    }
    std::cout<<"The whole size is "<<all_d.size()<<", in which "<<success_count<<" case success and "<<fail_count<<" cases fail"<<std::endl;
}


int main() {
    std::vector<ActItemThrift> all_d;
    scan_active_service(all_d);
    
    reverse(all_d.begin(), all_d.end());
    std::vector<std::thread> thread_batch_deal;
    int count_thread(8);
    size_t thread_deal_num(all_d.size()/count_thread);
    for (int i(0); i<count_thread; ++i) {
        size_t start = (i*thread_deal_num);
        size_t end = (i+1)*thread_deal_num;
        if (i == count_thread - 1) {
            end = all_d.size();
        }
        thread_batch_deal.push_back(std::thread(thread_deal, std::ref(all_d), start, end));
    }
    for (auto& s:thread_batch_deal) {
        s.join();
    }
    
}